package de.preissler.createview;

import java.util.LinkedList;
import java.util.List;

import me.prettyprint.cassandra.serializers.DoubleSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.RangeSlicesQuery;
import me.prettyprint.hector.api.query.SubColumnQuery;
import me.prettyprint.hector.api.query.SuperSliceQuery;

/**
 * Erstellt die materalsierte Sicht fuer den Join Qoutes und DAX
 * @author eric
 *
 */
public class JoinQuotesDAX
{
	private static final String DAX = "DAX";
	private static final String QUOTES = "Quotes";
	private static final String CLOSE = "close";
	private static final String KT = "Kt";
	private static final String BASE = "base";
	private static final String PI0 = "price at zero";
	private static final String QI0 = "#shares at zero";
	private static final String QIT = "#shares at t";
	private static final String FFI = "free float";
	private static final String CI = "correction";
	private static final String INDEX_ISIN = "DE0008469008";
	
	private Keyspace keyspace;
	
	public JoinQuotesDAX(Keyspace keyspace)
	{
		this.keyspace = keyspace;
	}
	
	/*
	 * Holt die ISINS
	 */
	private List<String> getISINs(Keyspace keyspace, String date)
	{
		SuperSliceQuery<String, String, String, Double> query = HFactory.createSuperSliceQuery(keyspace, StringSerializer.get(),
				StringSerializer.get(), StringSerializer.get(), DoubleSerializer.get());
		query.setColumnFamily(DAX);
		query.setKey(date);
		query.setRange(null, null, false, Integer.MAX_VALUE);
		List<HSuperColumn<String, String, Double>> result = query.execute().get().getSuperColumns();
		List<String> isins = new LinkedList<String>();
		
		for(HSuperColumn<String, String, Double> sc : result)
			isins.add(sc.getName());
		
		return isins;
	}
	
	/*
	 * Holt alle Schluessel raus
	 */
	private List<String> getKeys(Keyspace keyspace, String nameOfColumnFamily)
	{
		List<List<String>> listsOfISINs = new LinkedList<List<String>>();
		
		String start = null;
		String end = null;
		boolean firstLoop = true;
		//Max. werden nur 100 Schluessel zurueck gegeben, deswegen diese Schleife
		while(true)
		{
			List<String> isins = new LinkedList<String>();
			
	         RangeSlicesQuery<String, String, String> rangeSlicesQuery = 
	        	 HFactory.createRangeSlicesQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
		     rangeSlicesQuery.setColumnFamily(nameOfColumnFamily);
		     rangeSlicesQuery.setReturnKeysOnly();
		     rangeSlicesQuery.setKeys(start, end);
		     List<Row<String, String, String>> result = rangeSlicesQuery.execute().get().getList();
		     
		     for(int i = 0; i < result.size(); i++)
		     {
		    	 if((!firstLoop) && (i == 0))
		    	 {
		    		 
		    	 }
		    	 else
		    		 isins.add(result.get(i).getKey());
		    		 
		     }
		     
		     if(isins.isEmpty())
		    	 break;
		     else
		     {
		    	 start = isins.get(isins.size()-1);
		    	 listsOfISINs.add(isins);
		    	 firstLoop = false;
		     }
		}
		
		List<String> isins = new LinkedList<String>();
		for(List<String> subList : listsOfISINs)
		{
			isins.addAll(subList);
		}
	     
		return isins;
	}
	
	/*
	 * Erstellt den Index
	 */
	public void createIndex() throws Exception
	{
		//alle verfuegbaren ISINS und Dates holen
		List<String> dates = getKeys(keyspace, DAX);
		List<Exception> exceptions = new LinkedList<Exception>();
		Mutator<String> mutator = HFactory.createMutator(keyspace, MyUtil.SS);
		int i = 1;
		
		//Fuer jedes Datum
		for(String date : dates)
		{
			MyUtil.trace("Working on " + date + " (" + i + "/" + dates.size() + ")");
			List<String> isins = getISINs(keyspace, date);
			for(String isin : isins)
			{
				try
				{
					//Wenn es der DAX ist brauchen wir andere Werte
					if(!isin.equals(INDEX_ISIN))
					{
						//hier werden die Werte geholt
						double close = getValue(keyspace, QUOTES, isin, date, CLOSE);
						double pi0 = getValue(keyspace, DAX, date, isin, PI0);
						double qi0 = getValue(keyspace, DAX, date, isin, QI0);
						double qit = getValue(keyspace, DAX, date, isin, QIT);
						double ffit = getValue(keyspace, DAX, date, isin, FFI);
						double ci = getValue(keyspace, DAX, date, isin, CI);
						
						//Hier werden sie gespeichert
						saveRecord(mutator, date, isin, close, pi0, qi0, qit, ffit, ci);
					}
					else
					{
						//DAX Werte auslesen
						double kt = getValue(keyspace, DAX, date, isin, KT);
						double base = getValue(keyspace, DAX, date, isin, BASE);
						
						//DAX Werte speichern
						List<HColumn<String, Double>> columns = new LinkedList<HColumn<String, Double>>();
						columns.add(HFactory.createColumn(KT, kt, MyUtil.SS, MyUtil.DS));
						columns.add(HFactory.createColumn(BASE, base, MyUtil.SS, MyUtil.DS));
						mutator.insert(date, Main.SCF_RVT_JOIN, HFactory.createSuperColumn(isin, columns, MyUtil.SS, MyUtil.SS, MyUtil.DS));
					}
				}
				catch(Exception e)
				{
					exceptions.add(e);
				}
			}
			i++;
		}
		
		MyUtil.trace("Exceptions: " + exceptions.size());
	}
	
	private void saveRecord(Mutator<String> mutator,
			String date, String isin, double close, double pi0, double qi0,
			double qit, double ffit, double ci)
	{
		List<HColumn<String, Double>> columns = new LinkedList<HColumn<String, Double>>();
		
		columns.add(HFactory.createColumn(CLOSE, close, MyUtil.SS, MyUtil.DS));
		columns.add(HFactory.createColumn(PI0, pi0, MyUtil.SS, MyUtil.DS));
		columns.add(HFactory.createColumn(QI0, qi0, MyUtil.SS, MyUtil.DS));
		columns.add(HFactory.createColumn(QIT, qit, MyUtil.SS, MyUtil.DS));
		columns.add(HFactory.createColumn(FFI, ffit, MyUtil.SS, MyUtil.DS));
		columns.add(HFactory.createColumn(CI, ci, MyUtil.SS, MyUtil.DS));
		
		mutator.insert(date, Main.SCF_RVT_JOIN, HFactory.createSuperColumn(isin, columns, MyUtil.SS, MyUtil.SS, MyUtil.DS));
	}
	
	private double getValue(Keyspace keyspace, String scf, String key, String scn, String cn)
	{
		SubColumnQuery<String, String, String, Double> query = HFactory.createSubColumnQuery(keyspace, MyUtil.SS, MyUtil.SS, MyUtil.SS, MyUtil.DS);
		query.setColumnFamily(scf);
		query.setKey(key);
		query.setSuperColumn(scn);
		query.setColumn(cn);
		return query.execute().get().getValue();
	}
}
